package springboot_kafka_storm.storm;

import java.util.Iterator;
import java.util.List;
import java.util.Map;

import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Tuple;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.alibaba.fastjson.JSON;

import springboot_kafka_storm.entity.InfoGet;
import springboot_kafka_storm.service.InfoGetService;
import springboot_kafka_storm.util.GetSpringBean;


public class Bolt extends BaseRichBolt{

		private static final long serialVersionUID = 6542256546124282695L;
		private static final Logger logger = LoggerFactory.getLogger(Bolt.class);
		private InfoGetService infoGetService; 
		public  final static String FIELD="insert";
		
		@SuppressWarnings("rawtypes")
		@Override
		//prepare方法和Spout中的open方法类似，为Bolt提供了OutputCollector，用来从Bolt中发送Tuple
		public void prepare(Map map, TopologyContext arg1, OutputCollector collector) {
			infoGetService=GetSpringBean.getBean(InfoGetService.class);
		}
	  
		@Override
		//execute（）方法：是Bolt实现的核心，每次Bolt从流接收一个订阅的tuple，都会调用这个方法
		public void execute(Tuple tuple) {
			String msg=tuple.getStringByField(FIELD);//获取spout传来的数据，指定从spout的Constants.FIELD拿数据
			System.out.println("Bolt获取到的值"+msg);
			try{
				List<InfoGet> listUser =JSON.parseArray(msg,InfoGet.class);
				System.out.println(listUser);
				//移除age小于10的数据
				if(listUser!=null&&listUser.size()>0){
					Iterator<InfoGet> iterator = listUser.iterator();//定义迭代器
					 while (iterator.hasNext()) {
						 InfoGet user = iterator.next();
						 System.out.println(user);
						 if (user.getAge()<10) {
							 logger.warn("Bolt移除的数据:{}",user);
							 iterator.remove();
						 }
					 }
					if(listUser!=null&&listUser.size()>0){
						infoGetService.insertService(listUser);
					}
				}
			}catch(Exception e){
				logger.error("Bolt的数据处理失败!数据:{}",msg,e);
			}
		}

		@Override
		//cleanup()方法：用于释放Bolt占用的资源。Storm在终止一个Bolt之前会调用这个方法
		public void cleanup() {
		}

		@Override
		//declareOutputFields（）方法：用于声明当前Bolt发送的Tuple中包含的字段
		public void declareOutputFields(OutputFieldsDeclarer arg0) {
				
		}
}
